package framework;

import java.util.ArrayList;
import java.util.List;

/**
 * 
 * This filter merges N input pipes and sorts them by time when passing it to
 * the output pipe.
 * 
 */
public class AscTimeSorterAndMerger extends FilterFramework {

	/**
	 * The number of input pipes
	 */
	private int n;
	/**
	 * the number of open input pipes
	 */
	int numberOfInputsOpen;

	/**
	 * 
	 * Inner class representing a frame structure measurement id+field
	 * 
	 */
	private class FrameStructure {

		int id;
		long measure;
	}

	/**
	 * List of inputFrames on memory
	 */
	private List<List<FrameStructure>> inputFrames;

	/**
	 * Constructor of the AscTimeSorterAndMerger class
	 * 
	 * @param n The number of input pipes
	 */
	public AscTimeSorterAndMerger(int n) {
		super(n, 1);
		this.n = n;
		inputFrames = new ArrayList<List<FrameStructure>>(n);
		this.numberOfInputsOpen = n;
	}

	/**
	 * The main method of the filter. Starts when the thread is launched
	 */
	@Override
	public void run() {
		// Initially fill the inputFrames array with one frame from each input
		// pipe
		for (int i = 0; i < n; i++) {
			try {

				// since to know that a frame has ended the filter has to read
				// at least one id from the next frame (the id=0),
				// the readFrame() function will actually return and expect
				// frames without the first id (only the id,
				// the measurement itself will be there)
				// So this in the beginning is to clear from the pipe everything
				// until the first measurement in the frame
				while (this.ReadInt(i) != 0) {
					this.ReadLong(i);
				}

				// now we actually read the frame and put it in the
				// corresponding position in the list of frames
				// index in the list will be the same as the index of the port
				// from where it was read
				List<FrameStructure> frame = this.readFrame(i);
				inputFrames.add(i, frame);

			} catch (EndOfStreamException e) {
				// CloseInputPort(i);

				// one of the pipes has no more data
				System.out
						.print("\n"
								+ this.getName()
								+ "::AscTimeSorterAndMerger End of stream on input port "
								+ i + " - closing it;");
				numberOfInputsOpen--;
				break;
			}
		}

		// and read new one from the respective input pipe
		// (the pipe where the value sent to output was read)
		while (this.numberOfInputsOpen > 0) {
			// get smallest (oldest time) frame index on the array
			int index = this.getLowerTimeIndexFromFrameStructures();

			// get element
			List<FrameStructure> frame = inputFrames.remove(index);
			// send frame to output pipe
			this.sendFrameToOutput(frame);
			try {
				// read frame from input pipe and store it in the structure
				frame = this.readFrame(index);
				inputFrames.add(index, frame);
			} catch (EndOfStreamException e) {
				// this.CloseInputPort(index);

				// if the pipe is already closed, the corresponding position in
				// the frames list will be null,
				// so that no more attempts to read are tried
				inputFrames.add(index, null);
				numberOfInputsOpen--;
			}
		}
		// CloseOutputPort(0);
		ClosePorts();
	}

	/**
	 * This method write a frame to the output port
	 * @param frame The list of the frames to output. Each frame is the set of measurements.
	 */
	private void sendFrameToOutput(List<FrameStructure> frame) {
		for (FrameStructure struc : frame) {
			writeInt(struc.id, 0);
			writeLong(struc.measure, 0);
		}
	}

	/**
	 * This method returns the index in the inputFrames list which has the 'lowest' value -
	 * the lowest value of time. 
	 * 
	 * @return the index (corresponds to the input pipe index) of the lower time value
	 * in the frame structure. Or null, if no value is found.
	 */
	private Integer getLowerTimeIndexFromFrameStructures() {

		long minimum = Long.MAX_VALUE;
		Integer minLocation = null;

		for (int i = 0; i < inputFrames.size(); i++) {
			List<FrameStructure> frame = inputFrames.get(i);

			if (frame != null) {
				FrameStructure tempStruc = frame.get(Constants.TIME_ID);
				if (tempStruc.measure < minimum) {
					minimum = tempStruc.measure;
					minLocation = i;
				}
			}
		}

		return minLocation;
	}

	/**
	 * Reads one frame from the input pipe.
	 * 
	 * @param port The input pipe to read
	 * @return A list of ids and measurements representing a frame
	 * @throws EndOfStreamException
	 *             input pipe closed or no more data in input pipe
	 *             has to put the first Id=0 part
	 */
	private List<FrameStructure> readFrame(int port)
			throws EndOfStreamException {

		List<FrameStructure> frame = new ArrayList<FrameStructure>();

		Long firstValue = ReadLong(port);

		FrameStructure struc = new FrameStructure();
		struc.id = 0;
		struc.measure = firstValue;
		frame.add(struc);

		try {
			while (true) {

				struc = new FrameStructure();

				int id = ReadInt(port);

				if (id == 0) {
					break;
				}

				struc.id = id;
				struc.measure = ReadLong(port);

				frame.add(struc);
			}
		} catch (EndOfStreamException e) {
			return frame;
		}

		return frame;
	}
}
